package com.beiding.jsonanalyzer.analyze;

import com.alibaba.fastjson.JSONObject;
import com.beiding.jsonanalyzer.analyze.func.FuncBase;
import com.beiding.jsonanalyzer.analyze.query.ComparatorResult;
import com.beiding.jsonanalyzer.flow.BranchNode;
import com.beiding.jsonanalyzer.flow.ValueNode;
import com.beiding.jsonanalyzer.flow.ValueNodeFactory;
import com.beiding.jsonanalyzer.flow.flowevents.BatchEndEvent;
import com.beiding.jsonanalyzer.flow.flowevents.BatchStartEvent;
import com.beiding.jsonanalyzer.flow.flowevents.DataNodeEvent;
import com.beiding.jsonanalyzer.querylink.ComputeNodeUtils;

import java.util.*;
import java.util.logging.Logger;

/**
 * @author 丁常磊
 * @date 2022/4/24 15:09
 */
public class Joiner {

    private static Logger logger = Logger.getLogger(Joiner.class.getName());

    private static void putAllToMap(Map<String, Object> map, Object value) {
        if (value instanceof Map) {
            //左侧节点
            Map<?, ?> lv = (Map) value;
            lv.forEach((k, v) -> {
                if (k instanceof String) {
                    map.put((String) k, v);
                }
            });
        }
    }

    public static <W extends ValueNode> void leftJoin(BranchNode<W> r, BranchNode<W> left, BranchNode<W> right, FuncBase<ComparatorResult> comparator) {

        ValueNodeFactory<W> valueNodeFactory = r.getBranch().getValueNodeFactory();

        Map<Object, List<W>> leftMap = new HashMap<>();
        Map<Object, List<W>> rightMap = new HashMap<>();

        Map<Object, Set<W>> leftSetMap = new HashMap<>();

        Bre<W> bre = new Bre<W>(r, left, right) {

            @Override
            void onBatchStart(Object batch) {
                leftMap.put(batch, new ArrayList<>());
                rightMap.put(batch, new ArrayList<>());
                leftSetMap.put(batch, new LinkedHashSet<>());
                super.onBatchStart(batch);
            }

            @Override
            void onBatchEnd(Object batch) {
                List<W> rml = leftMap.remove(batch);
                List<W> rmr = rightMap.remove(batch);
                Set<W> lset = leftSetMap.remove(batch);
                executeOneSide(lset, r);
                super.onBatchEnd(batch);
            }

            @Override
            void onLeftData(Object batch, W data) {

                List<W> leftList = leftMap.get(batch);
                List<W> rightList = rightMap.get(batch);
                Set<W> leftSet = leftSetMap.get(batch);

                leftList.add(data);

                //执行内联
                boolean have = execInnerJoin(data, rightList, comparator, r);

                //如果没有找到匹配的元素,就加入单一元素进行专门处理
                if (!have) {
                    leftSet.add(data);
                }
            }

            @Override
            void onRightData(Object batch, W data) {
                List<W> leftList = leftMap.get(batch);
                List<W> rightList = rightMap.get(batch);
                Set<W> leftSet = leftSetMap.get(batch);

                rightList.add(data);


                //执行时,如果右侧找到了左侧的节点就将其移除掉
                execInnerJoin(data, leftList, comparator, r, leftSet);
            }
        };

        bre.init();

    }


    private static <W extends ValueNode> boolean execInnerJoin(W current, List<W> list, FuncBase<ComparatorResult> comparator, BranchNode<W> dataFlow, Set<W>... removeSet) {

        boolean hv = false;

        for (W node : list) {
            JSONObject jsonObject = new JSONObject();
            putAllToMap(jsonObject, current.get());
            putAllToMap(jsonObject, node.get());


            W w = dataFlow.getBranch().getValueNodeFactory().create();
            w.set(jsonObject);
            if (comparator.executeOne(w) == ComparatorResult.TRUE) {
                hv = true;

                //表示已经找到了匹配的元素,不需要单独列出
                for (Set<W> set : removeSet) {
                    set.remove(node);
                }
                dataFlow.signDataEvent(new DataNodeEvent<>(w));
            }
        }
        return hv;
    }

    public static <W extends ValueNode> void join(BranchNode<W> r, BranchNode<W> left, BranchNode<W> right, FuncBase<ComparatorResult> comparator) {

        Map<Object, List<W>> leftMap = new HashMap<>();
        Map<Object, List<W>> rightMap = new HashMap<>();

        Bre<W> bre = new Bre<W>(r, left, right) {

            @Override
            void onBatchStart(Object batch) {
                leftMap.put(batch, new ArrayList<>());
                rightMap.put(batch, new ArrayList<>());
                super.onBatchStart(batch);
            }

            @Override
            void onBatchEnd(Object batch) {
                List<W> rml = leftMap.remove(batch);
                List<W> rmr = rightMap.remove(batch);
                super.onBatchEnd(batch);
            }

            @Override
            void onLeftData(Object batch, W data) {

                List<W> leftList = leftMap.get(batch);
                List<W> rightList = rightMap.get(batch);
                leftList.add(data);

                //执行内联
                execInnerJoin(data, rightList, comparator, r);

            }

            @Override
            void onRightData(Object batch, W data) {
                List<W> leftList = leftMap.get(batch);
                List<W> rightList = rightMap.get(batch);
                rightList.add(data);

                //执行时,如果右侧找到了左侧的节点就将其移除掉
                execInnerJoin(data, leftList, comparator, r);
            }
        };

        bre.init();
    }

    /**
     * 执行时保证当前节点存在
     *
     * @param dataFlow
     */
    private static <W extends ValueNode> void executeOneSide(Set<W> set, BranchNode<W> dataFlow) {

        for (W current : set) {
            JSONObject jsonObject = new JSONObject();
            putAllToMap(jsonObject, current.get());
            dataFlow.signDataEvent(new DataNodeEvent<>(current));
        }

    }


    public static <W extends ValueNode> void outerJoin(BranchNode<W> r, BranchNode<W> left, BranchNode<W> right, FuncBase<ComparatorResult> comparator) {
        Map<Object, List<W>> leftMap = new HashMap<>();
        Map<Object, List<W>> rightMap = new HashMap<>();

        Map<Object, Set<W>> leftSetMap = new HashMap<>();
        Map<Object, Set<W>> rightSetMap = new HashMap<>();

        Bre<W> bre = new Bre<W>(r, left, right) {

            @Override
            void onBatchStart(Object batch) {
                leftMap.put(batch, new ArrayList<>());
                rightMap.put(batch, new ArrayList<>());
                leftSetMap.put(batch, new LinkedHashSet<>());
                rightSetMap.put(batch, new LinkedHashSet<>());
                super.onBatchStart(batch);
            }

            @Override
            void onBatchEnd(Object batch) {
                List<W> rml = leftMap.remove(batch);
                List<W> rmr = rightMap.remove(batch);
                Set<W> lset = leftSetMap.remove(batch);
                executeOneSide(lset, r);
                Set<W> rset = rightSetMap.remove(batch);
                executeOneSide(rset, r);
                super.onBatchEnd(batch);
            }

            @Override
            void onLeftData(Object batch, W data) {

                List<W> leftList = leftMap.get(batch);
                List<W> rightList = rightMap.get(batch);

                Set<W> leftSet = leftSetMap.get(batch);
                Set<W> rightSet = rightSetMap.get(batch);

                leftList.add(data);

                //执行内联
                boolean have = execInnerJoin(data, rightList, comparator, r, rightSet);

                //如果没有找到匹配的元素,就加入单一元素进行专门处理
                if (!have) {
                    leftSet.add(data);
                }
            }

            @Override
            void onRightData(Object batch, W data) {
                List<W> leftList = leftMap.get(batch);
                List<W> rightList = rightMap.get(batch);

                Set<W> leftSet = leftSetMap.get(batch);
                Set<W> rightSet = rightSetMap.get(batch);

                rightList.add(data);


                //执行时,如果右侧找到了左侧的节点就将其移除掉
                boolean have = execInnerJoin(data, leftList, comparator, r, leftSet);
                if (!have) {
                    rightSet.add(data);
                }
            }
        };

        bre.init();
    }


    //rightJoin 用leftJoin实现

    public static <W extends ValueNode> void rightJoin(BranchNode<W> r, BranchNode<W> left, BranchNode<W> right, FuncBase<ComparatorResult> comparator) {
        leftJoin(r, right, left, comparator);
    }

    public static <W> void union(BranchNode<W> r, BranchNode<W> left, BranchNode<W> right, boolean all) {

        if (all) {

            //创建一个分支控制
            Bre<W> bre = new Bre<W>(r, left, right) {
                @Override
                void onLeftData(Object batch, W data) {
                    r.signDataEvent(new DataNodeEvent<>(data));
                }

                @Override
                void onRightData(Object batch, W data) {
                    r.signDataEvent(new DataNodeEvent<>(data));
                }
            };

            //创建监听事件
            bre.init();

        } else {

            Map<Object, Set<W>> bm = new HashMap<>();

            //创建一个分支控制
            Bre<W> bre = new Bre<W>(r, left, right) {

                @Override
                void onBatchStart(Object batch) {
                    bm.put(batch, new LinkedHashSet<>());
                    super.onBatchStart(batch);
                }

                @Override
                void onBatchEnd(Object batch) {
                    bm.remove(batch);
                    super.onBatchEnd(batch);
                }

                @Override
                void onLeftData(Object batch, W data) {
                    Set<W> ws = bm.get(batch);
                    if (!ws.contains(data)) {
                        r.signDataEvent(new DataNodeEvent<>(data));
                    }
                }

                @Override
                void onRightData(Object batch, W data) {
                    Set<W> ws = bm.get(batch);
                    if (!ws.contains(data)) {
                        r.signDataEvent(new DataNodeEvent<>(data));
                    }
                }
            };

            //创建监听事件
            bre.init();

        }

    }

    private static class BatchHolder<W> {
        /**
         * 缓存的数据
         */
        List<W> dataCache;

        /**
         * 另一方的批对象
         */
        BatchHolder<W> another;

        Object batchObj;

        boolean end;


    }

    /**
     * 两侧分支控制
     */
    private static abstract class Bre<W> {


        private BranchNode<W> left;
        private BranchNode<W> right;

        private BranchNode<W> result;


        public Bre(BranchNode<W> result, BranchNode<W> left, BranchNode<W> right) {
            this.result = result;
            this.left = left;
            this.right = right;
        }

        /**
         * 用于批暂存
         */
        private Map<Object, BatchHolder> leftCacheBatchMap = new HashMap<>();

        /**
         * 批暂存
         */
        private Map<Object, BatchHolder> rightCacheBatchMap = new HashMap<>();

        /**
         * 当前左侧的批
         */
        private Object currentLeftBatch;

        /**
         * 当前右侧的批
         */
        private Object currentRightBatch;


        void onBatchStart(Object batch) {
            //发送批开始信号
            result.signDataEvent(new BatchStartEvent<>(batch));
        }

        /**
         * 左侧产生数据
         *
         * @param data
         */
        abstract void onLeftData(Object batch, W data);

        /**
         * 右侧产生数据
         *
         * @param data
         */
        abstract void onRightData(Object batch, W data);

        void onBatchEnd(Object batch) {
            //发送i结束信号
            result.signDataEvent(new BatchEndEvent<>(batch));
        }

        void init() {

            //刷新事件的处理

            ComputeNodeUtils.bindBaseEvent(left, result);
            ComputeNodeUtils.bindBaseEvent(right, result);

            //左侧批开始
            left.onDataEvent(BatchStartEvent.eventType(), event -> {
                Object batch = event.getBatch();
                this.currentLeftBatch = batch;
                //批对象不可重复,如果批对象重复. 覆盖并打印批失败信息
                BatchHolder<W> batchHolder = new BatchHolder<>();
                batchHolder.dataCache = new ArrayList<>();
                batchHolder.end = false;
                batchHolder.batchObj = batch;
                batchHolder.another = this.rightCacheBatchMap.get(batch);
                BatchHolder holder = this.leftCacheBatchMap.put(batch, batchHolder);

                if (holder != null) {
                    logger.warning("批对象重复:" + batch);
                }

                if (batchHolder.another != null) {
                    onBatchStart(batch);

                    //检查另一侧分支上是否有暂存的数据. 如果有,消费干净
                    if (batchHolder.another.dataCache.size() > 0) {
                        for (W w : batchHolder.another.dataCache) {
                            onRightData(batchHolder.another.batchObj, w);
                        }
                    }

                }

            });

            //右侧批开始
            right.onDataEvent(BatchStartEvent.eventType(), event -> {
                Object batch = event.getBatch();
                this.currentRightBatch = batch;
                //批对象不可重复,如果批对象重复. 覆盖并打印批失败信息
                BatchHolder<W> batchHolder = new BatchHolder<>();
                batchHolder.dataCache = new ArrayList<>();
                batchHolder.end = false;
                batchHolder.batchObj = batch;
                batchHolder.another = this.leftCacheBatchMap.get(batch);
                BatchHolder holder = this.rightCacheBatchMap.put(batch, batchHolder);

                if (holder != null) {
                    logger.warning("批对象重复:" + batch);
                }

                if (batchHolder.another != null) {
                    onBatchStart(batch);
                    //检查另一侧分支上是否有暂存的数据
                    if (batchHolder.another.dataCache.size() > 0) {
                        for (W w : batchHolder.another.dataCache) {
                            onLeftData(batchHolder.another.batchObj, w);
                        }
                    }
                }
            });


            //左侧数据事件
            left.onDataEvent(DataNodeEvent.eventType(), event -> {
                W dataNode = event.getData();

                //批相同
                if (currentLeftBatch == currentRightBatch) {
                    onLeftData(currentLeftBatch, dataNode);
                } else {
                    //如果批不同,等待右侧的批就绪
                    //暂时放入缓存
                    leftCacheBatchMap.get(currentLeftBatch).dataCache.add(dataNode);
                }

            });

            //右侧数据事件
            right.onDataEvent(DataNodeEvent.eventType(), event -> {
                W dataNode = event.getData();
                if (currentRightBatch == currentLeftBatch) {
                    onRightData(currentRightBatch, dataNode);
                } else {
                    rightCacheBatchMap.get(currentRightBatch).dataCache.add(dataNode);
                }
            });


            //左侧批结束
            left.onDataEvent(BatchEndEvent.eventType(), event -> {

                Object batch = event.getBatch();
                BatchHolder batchHolder = leftCacheBatchMap.get(batch);

                //检查是否结束
                batchHolder.end = true;
                if (batchHolder.another != null && batchHolder.another.end) {
                    leftCacheBatchMap.remove(batch);
                    rightCacheBatchMap.remove(batch);
                    onBatchEnd(batch);
                }
            });

            //右侧批开始
            right.onDataEvent(BatchEndEvent.eventType(), event -> {
                Object batch = event.getBatch();
                BatchHolder batchHolder = rightCacheBatchMap.get(batch);
                //检查是否结束
                batchHolder.end = true;
                if (batchHolder.another != null && batchHolder.another.end) {
                    leftCacheBatchMap.remove(batch);
                    rightCacheBatchMap.remove(batch);
                    onBatchEnd(batch);
                }
            });

        }

    }

}
